package com.zang.netty.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

/**
 *
 * 多路复用器 负责轮询多路复用器 Selctor
 * Multiplexer  /'mʌltɪ,plɛksɚ/  mao ti pu lak sir
 *
 * 1.使用构造方法设置初始参数、资源
 * 2.循环遍历selector去轮询
 * 3.根据新连接的SelectionKey通过ServerSocketChannel的accept接收客户端请求并创建SocketChannel实例
 * 4.解码请求流，打印消息，然后发送回执消息
 *
 * @author Zhang Qiang CodeBy 《Netty权威指南》第2版
 * @Date 2019/7/18 18:59
 */
public class MultiplexerTimeServer implements Runnable{

     /**
      * 多路复用器 构建选择器，提供选择已经就绪的任务的能力。
      * Selector会不断的轮询注册在其上的Channel，如果某个Channel发生读写等时间这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，进行后续IO操作
      * 一个Selector 可以同时轮询多个Channel，一个线程负责Selector的轮询就可以接入成千上万个客户端
      */
    private Selector selector;
     /**
      * 面向流的侦听套接字的通道 通过ServerSocketChannel.open()方法创建服务器套接字通道ServerSocketChannel
      */
    private ServerSocketChannel serverChannel;
     /**
      *  volatile特性
      *  当一个共享变量被volatile修饰时，它会保证修改的值立即被更新到主存。
      *  内存可见性：通俗来说就是，线程A对一个volatile变量的修改，对于其它线程来说是可见的，即线程每次获取volatile变量的值都是最新的。
      *  stop 用来设置是否关闭/停止
      */
    private volatile boolean stop;

     /**
      * 资源初始化，创建多路复用器Selector、ServerSocketChannel，对Channel和TCP参数进行配置
      */
    public MultiplexerTimeServer(int port){
        try {
            // 开启选择器 开启通道
            selector = Selector.open();
            serverChannel = ServerSocketChannel.open();

            // 绑定地址端口
            // InetSocketAddress： 绑定套接字的地址，本质上就是一套IP地址，或 null绑定到自动分配的套接字地址
            // backlog：待处理连接的最大数量
            serverChannel.socket().bind(new InetSocketAddress("127.0.0.1", port), 1024);
            // 设置异步非阻塞模式
            serverChannel.configureBlocking(false);
            // 注册地址 给定的选择器注册此频道，返回一个选择键（SelectionKey）
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);

            System.out.println(" The time server is start in port : " + port);

        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
     /**
      * Run!Run!Run!
      * 在While中循环遍历selector，休眠时间为1s。无论是否有读写时间发生都会每隔1s被唤醒一次
      *
      * selector.select()无参方法：当处于就绪状态的channel时，selector会返回该channel的selectionKey的集合。然后通过对就绪状态的channel集合进行迭代可以进行网络的异步读写
      */
    public void run() {
        int total = 0;
        while (!stop){
            try {
                total++;
                System.out.println("等待连接…… " + total);
                // 选择一组其相应通道准备好进行I/O操作的键，阻塞到至少有一个通道在你注册的事件上就绪了。timeout：最长阻塞时间
                selector.select(1000);
                // SelectionKey 选择键，是一个抽象类，表示selectableChannel在Selector中注册的标识。每个Channel向Selector注册时，都将会创建一个selectionKey
                //选择键将Channel与Selector建立了关系，并维护了channel事件。
                //可以通过cancel方法取消键，取消的键不会立即从selector中移除，而是添加到cancelledKeys中，在下一次select操作时移除它。所以在调用某个key时，需要使用isValid进行校验。
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                // 迭代器
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                // 无限循环轮询准备就绪的key
                while (it.hasNext()){
                    key = it.next();
                    System.out.println(" key : " + key);
                    it.remove();
                    try {
                        //处理input
                        handleInput(key);
                    }catch (Exception e){
                        if (key != null){
                            //要求取消该密钥的通道与其选择器的注册。
                            key.cancel();
                            // key.channel() 返回创建此键的通道。
                            if (key.channel() != null){
                                key.channel().close();
                            }
                        }
                    }
                }
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
        // 多路复用器关闭后所有注册在上面的channel和Pipe等资源都会被自动去除注册并关闭，所以不需要重复释放资源
        if (selector != null){
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

     /**
      * 处理input（新接入的客户端请求消息），根据selectionKey的操作位进行判断即可知道网络事件的类型，通过ServerSocketChannel的accept接收客户端的连接请求并创建SocketChannel实例（相当于安慰昵称了TCP的三次握手）
      *
      */
    private void handleInput(SelectionKey key) throws IOException {
        //是否有效
        if(key.isValid()){
            //测试此密钥的通道是否已准备好接受新的套接字连接。
            //如果已准备好则开始等待新连接接入并且在接入后获取通道
            if (key.isAcceptable()){
                //获取ServerSocket通道
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                // 阻塞并返回获取的通道（等待接受与此频道套接字的连接），获取通道。无论此通道的阻塞模式如何，通过此方法返回的套接字通道（如果有）将处于阻塞模式。
                SocketChannel sc = ssc.accept();
                // 关闭阻塞 Socket通道可以在非阻塞模式下运行。这个陈述虽然简单却有着深远的含义。传统Java socket的阻塞性质曾经是Java程序可伸缩性的最重要制约之一。
                // 非阻塞I/O是许多复杂的、高性能的程序构建的基础。要把一个socket通道置于非阻塞模式，我们要依靠所有socket通道类的公有超级类：SelectableChannel。
                sc.configureBlocking(false);
                // register添加新连接
                sc.register(selector, SelectionKey.OP_READ);
            }

            // 是否可读
            // 用于读取客户端的请求消息。 首先创建一个ByteBuffer，由于我们实现无法得知客户端发送的码流大小，所以先开辟一个1MB的缓冲区。然后调用SocketChannel人read方法读取请求码流（这里的read根据之前被设置而是非阻塞的）。
            if (key.isReadable()){
                // 返回通道
                SocketChannel sc = (SocketChannel) key.channel();
                //开始读取，allocate()分配一个新的字节缓冲区。
                ByteBuffer readerBuffer = ByteBuffer.allocate(1024);
                //从该通道读取到给定缓冲区的字节序列。
                int readBytes = sc.read(readerBuffer);
                if (readBytes > 0){
                    //使缓冲区准备好新的通道写入或相对获取操作序列：它将缓冲区当前的限制（limit）设置为当前位置（position），然后将位置（position）设置为0，用于后续对缓冲区的读取操作。
                    //即 变读为写，一般在从Buffer读出数据前调用。
                    readerBuffer.flip();
                    //返回当前位置和限制之间的元素数。(此缓冲区中剩余的元素数 )
                    byte[] bytes = new byte[readerBuffer.remaining()];
                    readerBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println(" The time server receive order : " + body);
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
                    // 执行发送
                    doWrite(sc, currentTime);
                }else if (readBytes < 0){
                    key.cancel();
                    sc.close();
                }else {}
            }
        }
    }

     /**
      * 执行发送
      * 将应答消息异步发送给客户端
      *
      */
    private void doWrite(SocketChannel channel, String response) throws IOException {
        if (response != null && response.trim().length() > 0){
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            // 由于异步非阻塞，所以不能保证一次能发需要的字节数组发送完，会出现“写半包”的问题，这就需要不断轮询selector将没发送完的byteBuffer发送完毕，然后通过ByteBuffer的hasRemain()方法判断消息是否发送完成
            channel.write(writeBuffer);
        }
    }

    public void stop(){
        this.stop = true;
    }

}
